package com.storm.lesson.WorldCount;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

public class WordCountTopology {

	public static void main(String[] args) throws Exception {

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("spout", new MyRandomSentenceSpout(), 1);

		builder.setBolt("split", new MySplit(" "), 8).shuffleGrouping("spout");

		builder.setBolt("count", new WordCount(), 3).fieldsGrouping("split",
				new Fields("words"));

		builder.setBolt("sum", new SumBolt(), 1).shuffleGrouping("count");

		Config conf = new Config();

		conf.setDebug(true);

		if (args != null && args.length > 0) {

			conf.setNumWorkers(3);

			// 集群提交
			StormSubmitter.submitTopology(args[0], conf,
					builder.createTopology());
		} else {

			conf.setMaxTaskParallelism(3);

			// 本地提交
			LocalCluster cluster = new LocalCluster();

			cluster.submitTopology("word-count", conf, builder.createTopology());
		}
	}

}
